package com.sf.iisp.kafka.listener;


import com.sf.iisp.bind.biz.BindingBiz;
import com.sf.iisp.bind.biz.KafkaTraceLogBiz;
import com.sf.iisp.bind.biz.UnbindingBiz;
import com.sf.iisp.bind.domain.KafkaTraceLog;
import com.sf.iisp.bind.domain.ReceiveBindingInfo;
import com.sf.iisp.bind.enums.BindType;
import com.sf.kafka.check.util.JsonUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.annotation.Resource;
import java.util.List;
import java.util.concurrent.Executor;

/**
 * Created by qi on 16-7-6.
 */
public class SGSListener extends AbstractListener {

    private static final Logger LOGGER = LoggerFactory.getLogger(AbstractListener.class);

    private Executor executor;

    @Resource
    private KafkaTraceLogBiz kafkaTraceLogBiz;

    @Resource
    private BindingBiz bindingBiz;

    @Resource
    private UnbindingBiz unbindingBiz;

    @Override
    public void handle(List<String> msgs) {
        executor.execute(new Runnable() {
            @Override
            public void run() {
                KafkaTraceLog log = null;

                for (String msg : msgs) {
                    LOGGER.info("receive message = {}", msg);
                    try {
                        log = JsonUtil.readData(msg, KafkaTraceLog.class);
                        log.setContent(msg);

                        //保存报文日志
                        kafkaTraceLogBiz.saveTraceLog(log);

                        //根据消息类型，判断为绑定，解绑，查询
                        if (BindType.BIND.toString().equals(log.getMsgType())) {
                            bindingBiz.bind(msg);
                        } else if (BindType.UNBIND.toString().equals(log.getMsgType())) {
                            unbindingBiz.unbind(msg);
                        } else if (BindType.QUERY.toString().equals(log.getMsgType())) {
                            LOGGER.info("no query interface  implementation..........");
                        } else {
                            LOGGER.error("invalid message type = {}", log.getMsgType());
                        }
                    } catch (Exception e) {
                        LOGGER.error(e.getMessage(), e);
                    }
                }
            }
        });
    }

    public void setExecutor(Executor executor) {
        this.executor = executor;
    }
}
